package com.zc.core.defaults;

import com.zc.core.ProtocolSupport;
import com.zc.core.device.*;
import com.zc.core.enums.ErrorCode;
import com.zc.core.exception.DeviceOperationException;
import com.zc.core.message.*;
import com.zc.core.message.interceptor.DeviceMessageSenderInterceptor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;

import static com.zc.core.device.DeviceConfigKey.connectionServerId;

@Slf4j
public class DefaultDeviceMessageSender implements DeviceMessageSender {

    //设备操作代理,用于管理集群间设备指令发送
    private final DeviceOperationBroker handler;

    //设备操作接口,用于发送指令到设备,以及获取配置等相关信息
    private final DeviceOperator operator;

    //设备注册中心,用于统一管理设备以及产品的基本信息,缓存,进行设备指令下发等操作
    private final DeviceRegistry registry;

    private static final long DEFAULT_TIMEOUT = TimeUnit.SECONDS.toMillis(Integer.getInteger("zc.device.message.default-timeout", 10));

    @Setter
    @Getter
    private long defaultTimeout = DEFAULT_TIMEOUT;

    private final DeviceMessageSenderInterceptor globalInterceptor;

    public DefaultDeviceMessageSender(DeviceOperationBroker handler,
                                      DeviceOperator operator,
                                      DeviceRegistry registry,
                                      DeviceMessageSenderInterceptor interceptor) {
        this.handler = handler;
        this.operator = operator;
        this.registry = registry;
        this.globalInterceptor = interceptor;
    }

    /**
     * 发送一个支持回复的消息.
     * <p>
     * ⚠️: 请勿自己实现消息对象,而应该使用框架定义的3种消息.
     * ⚠️: 如果是异步消息,将直接返回<code>{"success":true,"code":"REQUEST_HANDLING"}</code>
     *
     * @param message 具体的消息对象
     * @param <R>     返回类型
     * @return 异步发送结果
     * @see com.zc.core.message.property.ReadPropertyMessage
     * @see com.zc.core.message.property.ReadPropertyMessageReply
     * @see com.zc.core.message.property.WritePropertyMessage
     * @see com.zc.core.message.property.WritePropertyMessageReply
     * @see com.zc.core.message.function.FunctionInvokeMessage
     * @see com.zc.core.message.function.FunctionInvokeMessageReply
     * @see ErrorCode#CLIENT_OFFLINE
     * @see ErrorCode#REQUEST_HANDLING
     * @see DeviceMessageSenderInterceptor
     */
    @Override
    public <R extends DeviceMessageReply> Flux<R> send(Publisher<RepayableDeviceMessage<R>> message) {
        return send(message, this::convertReply);
    }

    protected <T extends DeviceMessageReply> T convertReply(Message sent, Object reply) {
        if (reply instanceof ChildDeviceMessageReply) {
            if (!(sent instanceof ChildDeviceMessage)) {
                ChildDeviceMessageReply messageReply = ((ChildDeviceMessageReply) reply);
                if (!messageReply.isSuccess()) {
                    //如果是可识别的错误则直接抛出异常
                    ErrorCode.of(messageReply.getCode())
                             .map(DeviceOperationException::new)
                             .ifPresent(err -> {
                                 throw err;
                             });
                }
                if (messageReply.getChildDeviceMessage() == null) {
                    ErrorCode.of(messageReply.getCode())
                             .map(DeviceOperationException::new)
                             .ifPresent(err -> {
                                 throw err;
                             });
                    throw new DeviceOperationException(ErrorCode.NO_REPLY);
                }
                return convertReply(((ChildDeviceMessageReply) reply).getChildDeviceMessage());
            }
        }
        return convertReply(reply);
    }

    /**
     * 消息回复转换器
     * @param obj 消息
     * @param <T>
     * @return
     */
    @SuppressWarnings("all")
    protected <T extends DeviceMessage> T convertReply(Object obj) {
        DeviceMessage result = null;
        // 判断是否是回复的消息
        if (obj instanceof DeviceMessageReply) {
            DeviceMessageReply reply = ((DeviceMessageReply) obj);
            if (!reply.isSuccess()) {
                //如果是可识别的错误则直接抛出异常
                ErrorCode.of(reply.getCode())
                         .map(code -> {
                             String msg = reply.getHeader("errorMessage")
                                               .map(String::valueOf)
                                               .orElse(code.getText());
                             return new DeviceOperationException(code, msg);
                         })
                         .ifPresent(err -> {
                             throw err;
                         });
            }
            result = reply;
        } else if (obj instanceof DeviceMessage) {
            result = (DeviceMessage) obj;
        } else if (obj instanceof Map) {
            result = (DeviceMessage) MessageType.convertMessage(((Map) obj)).orElse(null);
        }
        if (result == null) {
            throw new DeviceOperationException(ErrorCode.SYSTEM_ERROR, new ClassCastException("can not cast " + obj + " to DeviceMessageReply"));
        }
        return (T) result;
    }

    private <R extends DeviceMessage> Flux<R> logReply(DeviceMessage msg, Flux<R> flux) {
        if (log.isDebugEnabled()) {
            return flux
                    .doOnNext(r -> log.debug(
                            "receive device[{}] message[{}]: {}",
                            operator.getDeviceId(),
                            r.getMessageId(), r))

                    .doOnComplete(() -> log.debug(
                            "complete receive device[{}] message[{}]",
                            operator.getDeviceId(),
                            msg.getMessageId()))

                    .doOnCancel(() -> log.debug(
                            "cancel receive device[{}] message[{}]",
                            operator.getDeviceId(),
                            msg.getMessageId()));
        }
        return flux;
    }

    /**
     * 发送消息并获取返回
     *
     * @param message 消息
     * @param <R>     回复类型
     * @return 异步发送结果
     * @see DeviceMessageSender#send(Publisher)
     */
    @Override
    public <R extends DeviceMessage> Flux<R> send(DeviceMessage message) {
        return send(Mono.just(message), this::convertReply);
    }

    /**
     * @return 刷新获取连接的服务id
     */
    private Mono<String> refreshAndGetConnectionServerId() {
        return Mono
                .defer(() -> operator
                        .refreshConfig(Collections.singleton(connectionServerId.getKey()))
                        .then(operator.getConnectionServerId()));
    }

    private ChildDeviceMessage createChildDeviceMessage(String parentId, DeviceMessage message) {

        ChildDeviceMessage children = new ChildDeviceMessage();
        children.setDeviceId(parentId);
        children.setMessageId(message.getMessageId());
        children.setTimestamp(message.getTimestamp());
        children.setChildDeviceId(operator.getDeviceId());
        children.setChildDeviceMessage(message);

        // https://github.com/jetlinks/jetlinks-pro/issues/19
        // 重新设备子设备消息头
        if (null != message.getHeaders()) {
            children.setHeaders(new ConcurrentHashMap<>(message.getHeaders()));
        }
        // 通过父设备转发消息设置 true 是
        message.addHeader(Headers.dispatchToParent, true);
        // 检查是否存在循环依赖
        children.validate();
        return children;
    }

    /**
     * 通过网关设备发送消息
     * @param parentId 网关设备id
     * @param message 发送的消息
     * @return
     */
    private Flux<DeviceMessage> sendToParentDevice(String parentId,
                                                   DeviceMessage message) {
        // 验证父id和当前设备id是否相同（父id和当前设备id不允许相同）
        if (parentId.equals(operator.getDeviceId())) {
            return Flux
                    .error(
                            new DeviceOperationException(ErrorCode.CYCLIC_DEPENDENCE, "validation.parent_id_and_id_can_not_be_same")
                    );
        }

        // 创建子设备消息
        ChildDeviceMessage children = createChildDeviceMessage(parentId, message);

        return registry
                // 获取设备操作接口
                .getDevice(parentId)
                // 不存在则抛出异常 （未知的父设备）
                .switchIfEmpty(Mono.error(() -> new DeviceOperationException(ErrorCode.UNKNOWN_PARENT_DEVICE)))
                .flatMapMany(parent -> parent
                        // 获取消息发送器
                        .messageSender()
                        // 发送消息
                        .send(Mono.just(children), resp -> this.convertReply(message, resp)))
                ;
    }

    /**
     * 发送消息并自定义返回结果转换器
     *
     * @param message      消息
     * @param replyMapping 消息回复转换器
     * @param <R>          回复类型
     * @return 异步发送结果
     * @see DeviceMessageSender#send(Publisher)
     */
    public <R extends DeviceMessage> Flux<R> send(Publisher<? extends DeviceMessage> message, Function<Object, R> replyMapping) {
        return Mono
                .zip(
                        // 当前设备连接的服务器ID
                        operator
                                // 获取服务id
                                .getConnectionServerId()
                                // 如果为空，刷新获取服务id
                                .switchIfEmpty(refreshAndGetConnectionServerId())
                                // 如果为空添加默认空字符串
                                .defaultIfEmpty(""),
                        // 拦截器
                        operator
                                // 获取此设备使用的协议支持
                                .getProtocol()
                                // 获取设备消息发送拦截器, 用于拦截发送消息的行为.
                                .flatMap(ProtocolSupport::getSenderInterceptor)
                                // 如果为空添加默认拦截器
                                .defaultIfEmpty(DeviceMessageSenderInterceptor.DO_NOTING),
                        // 网关设备id
                        operator
                                // 获取设备自身的配置，网关设备id
                                .getSelfConfig(DeviceConfigKey.parentGatewayId)
                                // 如果为空添加默认空字符串
                                .defaultIfEmpty("")
                )
                .flatMapMany(serverAndInterceptor -> {

                    // 设备消息发送拦截器,用于在消息发送端拦截处理消息.
                    // 可用于在一些自定义回复逻辑的场景
                    DeviceMessageSenderInterceptor interceptor = serverAndInterceptor
                            .getT2()
                            // 添加全局拦截器
                            .andThen(globalInterceptor);

                    //当前设备连接的服务器ID
                    String server = serverAndInterceptor.getT1();

                    //网关设备id
                    String parentGatewayId = serverAndInterceptor.getT3();

                    // 子设备发送消息
                    // 有上级网关设备则通过父级设备发送消息
                    if (StringUtils.isEmpty(server) && StringUtils.hasText(parentGatewayId)) {
                        return Flux
                                .from(message)
                                // 在消息发送前触发. 执行此方法后将使用返回值{@link DeviceMessage}进行发送到设备.
                                .flatMap(msg -> interceptor.preSend(operator, msg))
                                .flatMap(msg -> this
                                        // 通过网关设备发送消息
                                        .sendToParentDevice(parentGatewayId, msg)
                                        // 在消息发送后触发.这里发送后并不是真正的发送，其实只是构造了整个发送的逻辑流{@link Flux}(参数 reply),
                                        .as(flux -> interceptor.afterSent(operator, msg, interceptor.doSend(operator, msg, flux)))
                                )
                                .map(r -> (R) r);
                    }

                    // 网关设备或者直连设备发送消息
                    return Flux
                            .from(message)
                            // 在消息发送前触发. 执行此方法后将使用返回值{@link DeviceMessage}进行发送到设备.
                            .flatMap(msg -> interceptor.preSend(operator,msg))
                            .concatMap(msg -> Flux
                                    .defer(() -> {
                                        //缓存中没有serverId,说明当前设备并未连接到平台.
                                        if (StringUtils.isEmpty(server)) {
                                            return interceptor.afterSent(operator, msg, Flux.error(new DeviceOperationException(ErrorCode.CLIENT_OFFLINE)));
                                        }
                                        // 发送既不管(true)
                                        boolean forget = msg.getHeader(Headers.sendAndForget).orElse(false);
                                        //定义处理来自设备的回复.
                                        Flux<R> replyStream = forget
                                                ? Flux.empty()
                                                : handler
                                                //监听来自其他服务的回复
                                                .handleReply(msg.getDeviceId(),
                                                             msg.getMessageId(),
                                                             // 超时时间
                                                             Duration.ofMillis(msg.getHeader(Headers.timeout)
                                                                     // 消息头不存在超时时间则添加默认时间
                                                                     .orElse(defaultTimeout)))
                                                // 消息回复转换
                                                .map(replyMapping)
                                                // 错误并尝试回复
                                                .onErrorResume(DeviceOperationException.class, error -> {
                                                    // 如果离线重新检查状态
                                                    if (error.getCode() == ErrorCode.CLIENT_OFFLINE) {
                                                        //返回离线错误,重新检查状态,以矫正设备缓存的状态
                                                        return operator
                                                                .checkState()
                                                                .then(Mono.error(error));
                                                    }
                                                    return Mono.error(error);
                                                })
                                                .onErrorMap(TimeoutException.class, timeout -> new DeviceOperationException(ErrorCode.TIME_OUT, timeout))
                                                .as(flux -> this.logReply(msg, flux));

                                        //发送消息到设备连接的服务器
                                        return handler
                                                // 发送消息
                                                .send(server, Mono.just(msg))
                                                .defaultIfEmpty(-1)
                                                .flatMapMany(len -> {
                                                    //设备未连接到服务器
                                                    if (len == 0) {
                                                        //尝试发起状态检查,同步设备的真实状态
                                                        return operator
                                                                // 检查状态
                                                                .checkState()
                                                                .flatMapMany(state -> {
                                                                    // 不在线
                                                                    if (DeviceState.online != state) {
                                                                        return interceptor.afterSent(operator, msg, Flux.error(new DeviceOperationException(ErrorCode.CLIENT_OFFLINE)));
                                                                    }
                                                                    /*
                                                                      设备在线,但是serverId对应的服务没有监听处理消息
                                                                         1. 服务挂了
                                                                         2. 设备缓存的serverId不对
                                                                     */
                                                                    //尝试发送给父设备
                                                                    if (StringUtils.hasText(parentGatewayId)) {
                                                                        log.debug("Device [{}] Cached Server [{}] Not Available,Dispatch To Parent [{}]",
                                                                                  operator.getDeviceId(),
                                                                                  server,
                                                                                  parentGatewayId);

                                                                        return interceptor
                                                                                .afterSent(operator, msg, sendToParentDevice(parentGatewayId, msg))
                                                                                .map(r -> (R) r);
                                                                    }
                                                                    log.warn("Device [{}] Cached Server [{}] Not Available",
                                                                             operator.getDeviceId(),
                                                                             server);

                                                                    return interceptor.afterSent(operator, msg, Flux.error(new DeviceOperationException(ErrorCode.SERVER_NOT_AVAILABLE)));
                                                                });
                                                    }
                                                    // -1 设备离线
                                                    else if (len == -1) {
                                                        return interceptor.afterSent(operator, msg, Flux.error(new DeviceOperationException(ErrorCode.CLIENT_OFFLINE)));
                                                    }
                                                    log.debug("send device[{}] message complete", operator.getDeviceId());
                                                    // 消息监听回复 replyStream
                                                    return interceptor.afterSent(operator, msg, replyStream);
                                                })
                                                ;
                                    })
                                    .as(flux -> interceptor
                                            .doSend(operator, msg, flux.cast(DeviceMessage.class))
                                            .map(_resp -> (R) _resp)));
                });

    }

    /**
     * 发送{@link com.zc.core.message.function.FunctionInvokeMessage}消息更便捷的API
     *
     * @param function 要执行的功能
     * @return FunctionInvokeMessageSender
     * @see DeviceMessageSender#send(Publisher)
     * @see com.zc.core.message.function.FunctionInvokeMessage
     * @see FunctionInvokeMessageSender
     */
    @Override
    public FunctionInvokeMessageSender invokeFunction(String function) {
        return new DefaultFunctionInvokeMessageSender(operator, function);
    }

    /**
     * 发送{@link com.zc.core.message.property.ReadPropertyMessage}消息更便捷的API
     *
     * @param property 要获取的属性列表
     * @return ReadPropertyMessageSender
     * @see DeviceMessageSender#send(Publisher)
     * @see com.zc.core.message.property.ReadPropertyMessage
     * @see ReadPropertyMessageSender
     */
    @Override
    public ReadPropertyMessageSender readProperty(String... property) {
        return new DefaultReadPropertyMessageSender(operator)
                .read(property);
    }

    /**
     * 发送{@link com.zc.core.message.property.WritePropertyMessage}消息更便捷的API
     *
     * @return WritePropertyMessageSender
     * @see DeviceMessageSender#send(Publisher)
     * @see com.zc.core.message.property.WritePropertyMessage
     * @see WritePropertyMessageSender
     */
    @Override
    public WritePropertyMessageSender writeProperty() {
        return new DefaultWritePropertyMessageSender(operator);
    }
}
